Python 操作 Redis 必备神器：redis-py 源码阅读

 △点击上方“Python猫”关注 ，回复“2”加入交流群

作者：肖恩

来源：游戏不存在

• redis协议规范

• redis-py概述

• redis-py基础使用

• RedisCommand

• Redis连接

• 连接池

• pipeline

• LuaScript

• lock

redis协议规范

RESP（Redis Serialization Protocol）是Redis客户端和服务端的通讯协议。数据示例如下:

1. +OK\r\n

2. -Error message\r\n

3. :1000\r\n

4. $6\r\nfoobar\r\n

5. *2\r\n$3\r\nfoo\r\n$3\r\nbar\r\n

6. *3\r\n:1\r\n:2\r\n:3\r\n

协议定义了5种类型:

1. +前缀表示字符串，后接字符串文本，以 \r\n结尾，通常用于命令结果

2. -前缀表示异常信息，后接以空格连接的两个字符串，以 \r\n结尾

3. :前缀表示整数，后接整数，以 \r\n结尾

4. $前缀表示定长的字符串，后接字符串长度， \r\n和字符串文本，以 \r\n结尾

5. *前缀表示数组，后接数组的长度和 \r\n，数组的每个元素可以由上面4种类型构成

协议还约定了Null等的实现，详情请看参考链接部分。下面示例了 LLEN mylist 的请求和响应

1. C: *2\r\n$4\r\nLLEN\r\n$6\r\nmylist\r\n

2. 
   

3. S: :48293\r\n

• 客户端发送了 LLEN mylist指令，指令序列化成RESP长度为2的数组，2个定长字符串分别是llen和mylist。

• 服务端响应整数48293，即mylist数据的长度。

Request-Response model是redis服务的请求响应模型，可以对比http协议的模式。redis服务端响应客户端的指令，处理后响应回复客户端，可以简单理解为一问一答。当然pipeline，pub/sub和monitor除外。

Redis-py 源码概述

本文使用的redis-py版本是 3.5.3， 文件及包信息是：

redis-py未依赖其它的包，代码量虽然不多，6000行左右，但是100%理解还是需要一定的时间和基础。本文从redis-py日常使用出发，也是redis-py的README中内容，介绍这些基础功能在源码中的实现。

redis-py基础使用

RedisCommand

redis-py的简单使用:

1. >>> import redis

2. >>> r = redis.Redis(host='localhost', port=6379, db=0)

3. >>> r.set('foo', 'bar')

4. True

5. >>> r.get('foo')

6. b'bar'

追踪redis-py的实现：

1. # client.py

2. 
   

3. class Redis(object)

4. 
   

5. def __init__(self, host='localhost', port=6379,

6. db=0, ..):

7. ...

8. connection_pool = ConnectionPool(**kwargs)

9. self.connection_pool = connection_pool

10. ...

11. 
    

12. def set(self, name, value, ex=None, px=None, nx=False, xx=False, keepttl=False)

13. ...

14. return self.execute_command('SET', *pieces)

15. 
    

16. def get(self, name):

17. return self.execute_command('GET', name)

18. 
    

19. # COMMAND EXECUTION AND PROTOCOL PARSING

20. def execute_command(self, *args, **options):

21. "Execute a command and return a parsed response"

22. conn = self.connection or pool.get_connection(command_name, **options)

23. conn.send_command(*args)

24. return self.parse_response(conn, command_name, **options)

注意：为了便于理解，示例代码和实际的代码有出入，省去了复杂的逻辑和异常等

• redis首先创造了一个到redis服务的连接，

• redis包装了redis的所有指令，使用命令模式执行指令。

• 执行命令就是使用创建的连接发送指令，然后解析和获取响应。这和redis协议上的Request-Response model行为一致。

Redis连接

继续查看连接的创建和执行:

1. # connection.py

2. class Connection(object)

3. 
   

4. def __init__(...):

5. self.host = host

6. self.port = int(port)

7. self._sock = connect()

8. 
   

9. def connect():

10. for res in socket.getaddrinfo(self.host, self.port, self.socket_type,

11. socket.SOCK_STREAM):

12. family, socktype, proto, canonname, socket_address = res

13. sock = socket.socket(family, socktype, proto)

14. ...

15. # TCP_NODELAY

16. sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)

17. # connect

18. sock.connect(socket_address)

19. ...

20. return sock

21. 
    

22. def pack_command(self, *args):

23. command = []

24. args = tuple(args[0].encode().split()) + args[1:]

25. ...

26. buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))

27. for arg in imap(self.encoder.encode, args):

28. buff = SYM_EMPTY.join(

29. (buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF))

30. output.append(buff)

31. output.append(arg)

32. ...

33. return command

34. 
    

35. def send_command(self, *args, **kwargs):

36. 
    

37. command = self.pack_command(args)

38. 
    

39. if isinstance(command, str):

40. command = [command]

41. for item in command:

42. self._sock.sendall(*args, **kwargs)

• connection维持了一个socket连接

• 收到redis的命令使用pack_command进行RESP的序列化打包

• 数据包使用socket发送

1. # connection.py

2. 
   

3. class Connection(object)

4. 
   

5. def __init__(...,parser_class=PythonParser,...);

6. self._parser = parser_class(socket_read_size=socket_read_size)

7. 
   

8. def read_response(self):

9. response = self._parser.read_response()

10. return response

11. 
    

12. class PythonParser(BaseParser):

13. "Plain Python parsing class"

14. def __init__(self, socket_read_size):

15. self.socket_read_size = socket_read_size

16. ...

17. self._sock = connection._sock

18. self._buffer = SocketBuffer(self._sock,

19. self.socket_read_size,

20. connection.socket_timeout)

21. self.encoder = connection.encoder

22. 
    

23. def read_response(self):

24. raw = self._buffer.readline()

25. 
    

26. byte, response = raw[:1], raw[1:]

27. 
    

28. # server returned an error

29. if byte == b'-':

30. response = nativestr(response)

31. ...

32. # single value

33. elif byte == b'+':

34. pass

35. # int value

36. elif byte == b':':

37. response = long(response)

38. # bulk response

39. elif byte == b'$':

40. length = int(response)

41. response = self._buffer.read(length)

42. # multi-bulk response

43. elif byte == b'*':

44. length = int(response)

45. response = [self.read_response() for i in xrange(length)]

46. if isinstance(response, bytes):

47. response = self.encoder.decode(response)

48. return response

• connection创建了一个parser用于读取和解析服务响应

• 默认的PythonParser使用SocketBuffer读取socket数据

• read_response实现了RESP协议的解析过程。对于每行数据 \r\n，第一个字符是响应类型，剩下的数据内容，如果是multi-bulk还需要循环读取多行。建议对比协议和发送请求进行详细阅读理解。

PythonParser是pure-python的实现，如果希望更高效，可以额外安装hiredis，会提供一个基于c的解析器 HiredisParser。

连接池

redis-py使用连接池来提高执行效率，主要的使用方法3个步骤，创建连接池，从连接池中获取有效连接执行命令，完成后释放连接，语句如下:

1. # redis.py

2. connection_pool = ConnectionPool(**kwargs)

3. pool.get_connection(command_name, **options)

4. 
   

5. try:

6. conn.send_command(*args)

7. ...

8. finally:

9. ...

10. pool.release(conn)

连接池一定要注意释放，可以用try/finally，也可以使用上下文装饰器，这里使用了前者

连接池的具体实现:

1. # connection.py

2. class ConnectionPool(object):

3. 
   

4. def __init__(...):

5. self._available_connections = []

6. self._in_use_connections = set()

7. 
   

8. def make_connection(self):

9. "Create a new connection"

10. return self.connection_class(**self.connection_kwargs)

11. 
    

12. def get_connection(self, command_name, *keys, **options)

13. try:

14. connection = self._available_connections.pop()

15. except IndexError:

16. connection = self.make_connection()

17. self._in_use_connections.add(connection)

18. ...

19. connection.connect()

20. return connection

21. 
    

22. def release(self, connection):

23. "Releases the connection back to the pool"

24. 
    

25. try:

26. self._in_use_connections.remove(connection)

27. except KeyError:

28. # Gracefully fail when a connection is returned to this pool

29. # that the pool doesn't actually own

30. pass

31. 
    

32. self._available_connections.append(connection)

• 连接池内部使用可用连接数组和正在使用连接集合管理所有连接

• 获取连接时候，优先从可用连接数组获取；没有可用连接会创建新的连接

• 所有获取到的连接会加入正在使用连接, 如果当前连接未连接会先建立连接

• 连接释放时会从正在使用连接集合中移除，然后加入可用连接数组数组，等待复用

到这里，我们基本理顺了一个redis指令执行的流程:

1. r = redis.Redis(host='localhost', port=6379, db=0)

2. r.set('foo', 'bar')

pipeline

redis还支持pipeline管线模式，可以批量发送一些命令，然后获取所有的结果:

1. >>> r = redis.Redis(...)

2. >>> pipe = r.pipeline()

3. >>> pipe.set('foo', 'bar').sadd('faz', 'baz').incr('auto_number').execute()

4. [True, True, 6]

pipeline的继承自redis，做了一些扩展

1. class Pipeline(Redis)

2. def __init__(...):

3. self.command_stack = []

4. 
   

5. def execute_command(self, *args, **kwargs):

6. self.command_stack.append((args, options))

7. return self

8. 
   

9. def execute(self, raise_on_error=True):

10. "Execute all the commands in the current pipeline"

11. stack = self.command_stack

12. 
    

13. execute = self._execute_pipeline

14. 
    

15. execute(conn, stack, raise_on_error)

16. 
    

17. def _execute_pipeline(self, connection, commands, raise_on_error):

18. # build up all commands into a single request to increase network perf

19. all_cmds = connection.pack_commands([args for args, _ in commands])

20. connection.send_packed_command(all_cmds)

21. 
    

22. response = []

23. for args, options in commands:

24. response.append(

25. self.parse_response(connection, args[0], **options))

26. 
    

27. return response

• pipeline使用一个stack来临时存储批量发送的命令，同时返回自身，这样可以支持链式语法

• execute时候才正式发送指令

• 发送指令后再依次获取服务响应，打包称一个数组统一返回

LuaScript

redis使用lua脚本来处理事务,使用方法如下:

1. >>> r = redis.Redis()

2. >>> lua = """

3. ... local value = redis.call('GET', KEYS[1])

4. ... value = tonumber(value)

5. ... return value * ARGV[1]"""

6. >>> multiply = r.register_script(lua)

7. >>> r.set('foo', 2)

8. >>> multiply(keys=['foo'], args=[5])

9. 10

• lua脚本中定义了KEYS和ARGV两个数组用于接受参数，KEY的第一个值(lua数组从1开始)是key的名称，ARGV的第一个值是倍数

• 脚本需要进行注册

• redis-py中把参数传递给脚本并执行得到结果

脚本的实现原理:

1. # client.py

2. class Redis(object):

3. 
   

4. def register_script(self, script):

5. return Script(self, script

6. 
   

7. def script_load(self, script):

8. "Load a Lua ``script`` into the script cache. Returns the SHA."

9. return self.execute_command('SCRIPT LOAD', script)

10. 
    

11. def evalsha(self, sha, numkeys, *keys_and_args):

12. return self.execute_command('EVALSHA', sha, numkeys, *keys_and_args)

13. 
    

14. class Script(object):

15. "An executable Lua script object returned by ``register_script``"

16. 
    

17. def __init__(self, registered_client, script):

18. self.registered_client = registered_client

19. self.script = script

20. # Precalculate and store the SHA1 hex digest of the script.

21. ...

22. self.sha = hashlib.sha1(script).hexdigest()

23. 
    

24. def __call__(self, keys=[], args=[], client=None):

25. "Execute the script, passing any required ``args``"

26. args = tuple(keys) + tuple(args)

27. # make sure the Redis server knows about the script

28. ...

29. try:

30. return client.evalsha(self.sha, len(keys), *args)

31. except NoScriptError:

32. # Maybe the client is pointed to a differnet server than the client

33. # that created this instance?

34. # Overwrite the sha just in case there was a discrepancy.

35. self.sha = client.script_load(self.script)

36. return client.evalsha(self.sha, len(keys), *args

• lua脚本通过 script load 加载到redis服务，并获得一个sha值，sha值可以重用，避免多次加载同一脚本

• 通过 evalsha 执行脚本

lock

redis-py还提供了一个全局锁的实现, 可以跨进程同步:

1. try:

2. with r.lock('my-lock-key', blocking_timeout=5) as lock:

3. # code you want executed only after the lock has been acquired

4. except LockError:

5. # the lock wasn't acquired

下面是其实实现:

1. # lock.py

2. class Lock(object):

3. 
   

4. LUA_RELEASE_SCRIPT = """

5. local token = redis.call('get', KEYS[1])

6. if not token or token ~= ARGV[1] then

7. return 0

8. end

9. redis.call('del', KEYS[1])

10. return 1

11. ""

12. 
    

13. def __init__(...):

14. ...

15. self.redis = redis

16. self.name = name

17. self.local = threading.local() if self.thread_local else dummy()

18. self.local.token = None

19. cls = self.__class__

20. cls.lua_release = client.register_script(cls.LUA_RELEASE_SCRIPT)

21. 
    

22. def __enter__(self):

23. # force blocking, as otherwise the user would have to check whether

24. # the lock was actually acquired or not.

25. if self.acquire(blocking=True):

26. return self

27. raise LockError("Unable to acquire lock within the time specified")

28. 
    

29. def __exit__(self, exc_type, exc_value, traceback):

30. self.release()

31. 
    

32. def acquire(self, blocking=None, blocking_timeout=None, token=None):

33. ...

34. token = uuid.uuid1().hex.encode()

35. self.redis.set(self.name, token, nx=True, px=timeout)

36. ...

37. self.local.token = token

38. ...

39. 
    

40. def release(self):

41. expected_token = self.local.token

42. self.local.token = None

43. self.lua_release(keys=[self.name],

44. args=[expected_token],

45. client=self.redis)

• LUA_RELEASE_SCRIPT使用lua脚本来处理删除token的事务

• lock使用线程变量来存储token值，保证多线程并发可以正常

• _enter和_exit是装饰器语法，保证可以合法的获取和释放

• 申请锁的时候获取一个临时的token，然后设置到redis服务中，这个token是有生命周期的，可以超时自动释放。

• 释放的时候清理线程本地变量和redis服务中的变量

TODO

源码中的 publish/subscibe ， Monitor , Sentinel 和事务等内容，个人认为并不在主线任务上，留待后续再行介绍。

参考链接

• https://redis.io/topics/protocol

• https://github.com/andymccurdy/redis-py

• https://pypi.org/project/hiredis/#description

分享与在看是对我最大的支持！